/*----------------------------------------------------------------------------------------
*
*  Copyright 2019, Gao Hai Hui, <fromzeropoint@126.com>.  All rights reserved.
*  https://gitee.com/helloworldghh/xoskit.git
*  Use of this source code is governed by a MIT license
*  that can be found in the License file.
*
----------------------------------------------------------------------------------------*/
#include "../import/head.h"
#include "../data_struct/head.h"
#include "../macro/head.h"
#include "aio_udp.h"
#include "head.h"

/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos_aio
{
    aio_udp::aio_udp( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
    }

    aio_udp::~aio_udp()
    {
    }

    ////////////////////////////////////////////////////////////////////////////////////
    // 

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nMsg[0]  : cmd type
    // 
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::user_init( xos_box::i_msg * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        i_connect * pUserData = (i_connect *)lpMsg->get_void( 0, 0 );
        xos::i_socket * lpSocket = 0;
        aio_key * lpKey = 0;
        int hSocket = -1;

        if( 0 == nRet )
        {
            lpMsg->set_msg( 0, i_aio::AIO_UDP_INIT_RET );
            lpMsg->set_bool( 0, false );
            lpMsg->set_void( 1, 0 );
        }

        if( 0 == nRet )
        {
            lpSocket = mgr::container()->xos()->socket();
        }

        if( 0 == nRet )
        {
            uint16_t usLocalPort = (uint16_t)lpMsg->get_int( 1, 0 );
            const char * pszLocalIp = lpMsg->get_str( 1, 0, 0, 0 );
            nRet = lpSocket->udp_init( pszLocalIp, usLocalPort );
        }

        if( 0 == nRet )
        {
            hSocket = (int)lpSocket->get_socket();
            lpSocket->set_async( true );
            uint16_t usLocalPort = 0;
            char szLocalIp[100] = { 0 };
            nRet = lpSocket->get_local_addr( szLocalIp, &usLocalPort );
            if( 0 == nRet )
            {
                lpMsg->set_str( 1, szLocalIp );
                lpMsg->set_int( 1, usLocalPort );
            }
        }

        if( 0 == nRet )
        {
            lpKey = aio_key::get_item_from_pool();
            lpKey->init();
            lpKey->m_lpUserKey = pUserData;
            lpKey->m_pSocket = lpSocket;
            lpKey->m_enumHandleType = aio_key::H_TYPE_UDP;
            lpKey->m_eState = aio_key::KEY_STATE_RUNNING;
            lpKey->m_bCanWrite = true;
        }

        if( 0 == nRet )
        {
            lpKey->m_nEpollFlags = EPOLLIN | EPOLLOUT | EPOLLET;
            struct epoll_event ev;
            uint64_t u = (uint64_t)lpKey;
            u |= ( 0x01 & lpKey->m_nInstance );
            ev.data.ptr = (void *)u;
            ev.events = lpKey->m_nEpollFlags;
            int ret = epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_ADD, hSocket, &ev );
            if( 0 != ret )
            {
                cout << "user_init_server epoll_ctl err = " << strerror( errno ) << endl;
            }
        }

        if( 0 == nRet )
        {
            lpMsg->set_void( 1, lpKey );
            lpMsg->set_bool( 0, true );
            lpSocket = 0;
            lpKey = 0;
        }

        m_pAioImpl->notify( lpMsg );
        lpMsg = 0;

        if( lpSocket )
        {
            lpSocket->release();
            lpSocket = 0;
        }

        if( lpKey )
        {
            lpKey->m_pSocket = 0;
            lpKey->release();
            lpKey = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nMsg[0]  : cmd type
    // 
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_nData[0] : send len
    // m_lpBuf[0] : send buf
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::user_send( xos_box::i_msg * lpMsg )
    {
        using namespace std;
        int nRet = 0;
        int ret = 0;

        aio_key * lpKey = (aio_key *)lpMsg->get_void( 1, 0 );

        lpMsg->set_msg( 0, i_aio::AIO_UDP_SEND_RET );
        lpMsg->set_bool( 0, false );
        lpMsg->set_int( 0, 0 );

        {
            aio_data * pData = aio_data::get_item_from_pool();
            pData->m_pMsg = lpMsg;
            lpKey->m_pend_send_list.push_back( pData );
        }

        while( ( 0 == ret ) && ( aio_key::KEY_STATE_RUNNING == lpKey->m_eState ) )
        {
            ret = udp_send( lpKey );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nMsg[0]  : cmd type
    // 
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_nData[0] : send len
    // m_lpBuf[0] : send buf
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::udp_send( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        xos::i_socket * lpSocket = lpKey->m_pSocket;
        xos_box::i_msg * lpMsg = 0;
        xos_box::i_buf * lpBuf = 0;
        int nResult = -1;

        if( ( 0 == nRet ) && ( aio_key::KEY_STATE_RUNNING != lpKey->m_eState ) )
        {
            nRet = 1;
        }

        if( ( 0 == nRet ) && !lpKey->m_bCanWrite )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            if( lpKey->m_pend_send_list.size() > 0 )
            {
                aio_data * pData = lpKey->m_pend_send_list.front();
                lpKey->m_pend_send_list.pop_front();
                lpMsg = pData->m_pMsg;
                pData->m_pMsg = 0;
                pData->release();
                pData = 0;
                lpBuf = lpMsg->get_buf( 0, 0 );
            }
            else
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            uint16_t usPort = (uint16_t)lpMsg->get_int( 2, 0 );
            const char * pszIp = lpMsg->get_str( 2, 0, 0, 0 );
            const char * pszData = lpBuf->get_data( 0, 0, 0, 0 );
            int len = lpBuf->get_len( 0 );
            int nSendLen = lpSocket->udp_send( pszData, len, pszIp, usPort );
            if( nSendLen > 0 )
            {
                if( nSendLen == len )
                {
                    lpMsg->set_bool( 0, true );
                }
                lpMsg->set_int( 0, nSendLen );
                nResult = 0;
            }
            else if( errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK )
            {
                lpKey->m_bCanWrite = false;
                nResult = 1;
                nRet = 1;
            }
            else
            {
                lpKey->m_bCanWrite = false;
                nResult = 2;
                nRet = 1;
            }
        }

        if( ( 0 == nResult ) || ( 2 == nResult ) )
        {
            m_pAioImpl->notify( lpMsg );
            lpMsg = 0;
        }

        if( lpMsg )
        {
            aio_data * p = aio_data::get_item_from_pool();
            p->init();
            p->m_pMsg = lpMsg;
            lpKey->m_pend_send_list.push_front( p );
            lpMsg = 0;
            p = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nMsg[0]  : cmd type
    // 
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_nData[0] : recv len
    // m_lpBuf[0] : recv buf
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::udp_recv( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        xos::i_socket * lpSocket = lpKey->m_pSocket;
        xos_box::i_msg * lpMsg = 0;
        xos_box::i_buf * lpBuf = 0;
        int nResult = -1;

        if( ( 0 == nRet ) && ( aio_key::KEY_STATE_RUNNING != lpKey->m_eState ) )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            lpBuf = mgr::container()->box()->buf();
            lpMsg = mgr::container()->box()->msg();
            lpMsg->set_msg( 0, i_aio::AIO_UDP_RECV_RET );
            lpMsg->set_bool( 0, false );
            lpMsg->set_void( 0, lpKey->m_lpUserKey );
            lpMsg->set_void( 1, lpKey );
            lpMsg->set_buf( 0, lpBuf );
            lpMsg->set_int( 0, 0 );
        }

        if( 0 == nRet )
        {
            char * pBuf = lpBuf->get_data( 0, 0, 0, 0 );
            int nSize = lpBuf->get_buf_size( 0 );
            char szIp[100] = { 0 };
            uint16_t usPort = 0;
            int len = lpSocket->udp_recv( pBuf, nSize, szIp, &usPort );
            if( len > 0 )
            {
                lpMsg->set_str( 2, szIp );
                lpMsg->set_int( 2, usPort );
                lpMsg->set_int( 0, len );
                lpMsg->set_bool( 0, true );
                nResult = 0;
            }
            else if( 0 == len )
            {
                nResult = 2;
            }
            else if( errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK )
            {
                nResult = 1;
                nRet = 1;
            }
            else
            {
                lpMsg->set_int( 0, -1 );
                nResult = 2;
                nRet = 1;
            }
        }

        if( ( 0 == nResult ) || ( 2 == nResult ) )
        {
            m_pAioImpl->notify( lpMsg );
            lpMsg = 0;
            lpBuf = 0;
        }

        if( lpMsg )
        {
            lpMsg->release();
            lpMsg = 0;
        }

        if( lpBuf )
        {
            lpBuf->release();
            lpBuf = 0;
        }

        return nRet;
    }

    // 
    // m_nMsg[0]  : cmd type
    // m_nData[1] : how
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::shut_down( xos_box::i_msg * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = (aio_key *)lpMsg->get_void( 1, 0 );
        xos::i_socket * lpSocket = lpKey->m_pSocket;
        int how = lpMsg->get_int( 1, 0 );

        {
            lpMsg->set_msg( 0, i_aio::AIO_UDP_SHUTDOWN_RET );
            lpMsg->set_bool( 0, true );
        }
        {
            lpSocket->shut_down( how );
        }
        {
            m_pAioImpl->notify( lpMsg );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nMsg[0]  : cmd type
    // m_nData[1] : linger.l_onoff
    // m_nData[2] : linger.l_linger
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_udp::close_net( xos_box::i_msg * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = (aio_key *)lpMsg->get_void( 1, 0 );
        {
            lpMsg->set_msg( 0, i_aio::AIO_UDP_CLOSE_RET );
            lpMsg->set_bool( 0, true );
        }
        {
            lpKey->m_eState = aio_key::KEY_STATE_CLOSING;
        }
        {
            aio_data * p = aio_data::get_item_from_pool();
            p->init();
            p->m_pMsg = lpMsg;
            m_pAioImpl->m_closing_list.push_back( p );
            lpMsg = 0;
            p = 0;
        }

        return nRet;
    }

} // xos_aio
